package org.alis.rpc.core;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import org.alis.rpc.common.bean.RpcRequest;
import org.alis.rpc.common.bean.RpcResponse;
import org.alis.rpc.common.contant.RpcConstants;
import org.alis.rpc.common.util.ServerUtils;
import org.alis.rpc.common.util.ThreadPoolUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cglib.reflect.FastClass;

import java.util.Map;

/**
 * RpcServerHandler desc:
 *
 * @author luochuan
 * @date 2020/10/9 15:11
 */
public class RpcServerHandler<V>  extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Logger logger = LoggerFactory.getLogger(RpcServerHandler.class);

    private final Map<String, V> handlerMap;

    public RpcServerHandler(Map<String, V> handlerMap) {
        this.handlerMap = handlerMap;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, final RpcRequest request) {
        // filter beat ping
        if (RpcConstants.BEAT_ID.equalsIgnoreCase(request.getRequestId())) {
            logger.info("Server read heartbeat ping");
            return;
        }
        ThreadPoolUtils.execute(()->{
            logger.info("Receive request " + request.getRequestId());
            RpcResponse<V> response = new RpcResponse<>();
            response.setRequestId(request.getRequestId());
            try {
                V result = handle(request);
                response.setResult(result);
            } catch (Throwable t) {
                response.setError(t.toString());
                logger.error("RPC Server handle request error", t);
            }
            ctx.writeAndFlush(response).addListener((ChannelFutureListener) channelFuture -> logger.info("Send response for request " + request.getRequestId()));

        });

    }

    @SuppressWarnings("all")
    private V handle(RpcRequest request) throws Throwable {
        String className = request.getClassName();
        String version = request.getVersion();
        String serviceKey = ServerUtils.makeServiceKey(className, version);
        V serviceBean = handlerMap.get(serviceKey);
        if (serviceBean == null) {
            logger.error("Can not find service implement with interface name: {} and version: {}", className, version);
            return null;
        }

        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();

        logger.debug(serviceClass.getName());
        logger.debug(methodName);
        for (Class<?> parameterType : parameterTypes) {
            logger.debug(parameterType.getName());
        }
        for (Object parameter : parameters) {
            logger.debug(parameter.toString());
        }

        FastClass serviceFastClass = FastClass.create(serviceClass);
        int methodIndex = serviceFastClass.getIndex(methodName, parameterTypes);
        return (V) serviceFastClass.invoke(methodIndex, serviceBean, parameters);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.warn("Server caught exception: " + cause.getMessage());
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            ctx.channel().close();
            logger.warn("Channel idle in last {} seconds, close it", RpcConstants.BEAT_TIMEOUT);
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}